package ex.learnflink;

import ex.datatypes.TaxiRide;
import ex.util.generator.TaxiRideGenerator;
import ex.util.GeoUtils;
import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.Collector;

/**
 * 自定义函数使用
 * see https://nightlies.apache.org/flink/flink-docs-master/zh/docs/learn-flink/etl/
 */
public class EtlExample01 {
    private final SourceFunction<TaxiRide> source;
    /** Creates a job using the source and sink provided. */
    public EtlExample01(SourceFunction<TaxiRide> source) {
        this.source = source;
    }
    public static void main(String[] args) throws Exception {
        EtlExample01 job=new EtlExample01(new TaxiRideGenerator());
        job.execute();
    }

    public JobExecutionResult execute() throws Exception {
        // set up streaming execution environment
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<TaxiRide> dataStream=  env.addSource(source);
        dataStream.print();

        DataStream<EnrichedRide> enrichedNYCRides1 = dataStream
                .filter(new RideCleansingSolution.NYCFilter())//自定义FilterFunction
                .map(new Enrichment());//自定义MapFunction,MapFunction 只适用于一对一的转换

        DataStream<EnrichedRide> enrichedNYCRides2 = dataStream
                .flatMap(new NYCEnrichment());//自定义FlatMapFunction,使用接口中提供的 Collector ，flatmap() 可以输出你想要的任意数量的元素，也可以一个都不发

        DataStream<EnrichedRide> enrichedNYCRides3 = dataStream
                 .flatMap(new NYCEnrichment())
//                .keyBy(enrichedRide -> enrichedRide.startCell);//对 startCell 进行 GROUP BY 再排序
                .keyBy(ride -> GeoUtils.mapToGridCell(ride.startLon, ride.startLat));//对 startCell 进行 GROUP BY 再排序



        enrichedNYCRides3.print();


        // run the pipeline and return the result
        return env.execute("Taxi Ride Cleansing");
    }
    public static class NYCEnrichment implements FlatMapFunction<TaxiRide, EnrichedRide> {

        @Override
        public void flatMap(TaxiRide taxiRide, Collector<EnrichedRide> out) throws Exception {
            FilterFunction<TaxiRide> valid = new RideCleansingSolution.NYCFilter();
            if (valid.filter(taxiRide)) {
                out.collect(new EnrichedRide(taxiRide));
            }
        }
    }
    public static class Enrichment implements MapFunction<TaxiRide, EnrichedRide> {

        @Override
        public EnrichedRide map(TaxiRide taxiRide) throws Exception {
            return new EnrichedRide(taxiRide);
        }
    }
    public static class EnrichedRide extends TaxiRide {
        public int startCell;
        public int endCell;

        public EnrichedRide() {}

        public EnrichedRide(TaxiRide ride) {
            this.rideId = rideId;
            this.isStart = isStart;
            this.eventTime = eventTime;
            this.startLon = startLon;
            this.startLat = startLat;
            this.endLon = endLon;
            this.endLat = endLat;
            this.passengerCnt = passengerCnt;
            this.taxiId = taxiId;
            this.driverId = driverId;
            this.startCell = GeoUtils.mapToGridCell(ride.startLon, ride.startLat);
            this.endCell = GeoUtils.mapToGridCell(ride.endLon, ride.endLat);
        }

        public String toString() {
            return super.toString() + ",==========" +
                    Integer.toString(this.startCell) + "," +
                    Integer.toString(this.endCell);
        }
    }
}
